package mq

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/ammyhaber/seckill/config"
	"github.com/streadway/amqp"
)

// RabbitMQ 消息队列客户端
type RabbitMQ struct {
	conn         *amqp.Connection
	channel      *amqp.Channel
	queueName    string
	exchange     string
	routingKey   string
	deadExchange string
	deadQueue    string
	expiration   int // 消息过期时间（秒）
}

// NewRabbitMQ 创建RabbitMQ实例
func NewRabbitMQ(queueName, exchange, routingKey string) (*RabbitMQ, error) {
	cfg := config.GetConfig()

	// 构建连接URL
	url := fmt.Sprintf("amqp://%s:%s@%s:%d/%s",
		cfg.RabbitMQ.Username,
		cfg.RabbitMQ.Password,
		cfg.RabbitMQ.Host,
		cfg.RabbitMQ.Port,
		cfg.RabbitMQ.VHost,
	)

	// 建立连接
	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, fmt.Errorf("连接RabbitMQ失败: %w", err)
	}

	// 创建通道
	channel, err := conn.Channel()
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("创建通道失败: %w", err)
	}

	// 创建RabbitMQ实例
	mq := &RabbitMQ{
		conn:         conn,
		channel:      channel,
		queueName:    queueName,
		exchange:     exchange,
		routingKey:   routingKey,
		deadExchange: "dead.exchange",
		deadQueue:    cfg.RabbitMQ.DeadLetterQueue,
		expiration:   cfg.RabbitMQ.OrderExpiration,
	}

	return mq, nil
}

// NewOrderMQ 创建订单消息队列
func NewOrderMQ() (*RabbitMQ, error) {
	cfg := config.GetConfig()
	return NewRabbitMQ(cfg.RabbitMQ.OrderQueue, "order.exchange", "order.routing")
}

// InitOrderQueue 初始化订单队列
func (r *RabbitMQ) InitOrderQueue() error {
	// 声明死信交换机
	err := r.channel.ExchangeDeclare(
		r.deadExchange, // 交换机名称
		"direct",       // 交换机类型
		true,           // 持久化
		false,          // 自动删除
		false,          // 内部使用
		false,          // 非阻塞
		nil,            // 额外参数
	)
	if err != nil {
		return fmt.Errorf("声明死信交换机失败: %w", err)
	}

	// 声明死信队列
	_, err = r.channel.QueueDeclare(
		r.deadQueue, // 队列名称
		true,        // 持久化
		false,       // 自动删除
		false,       // 排他性
		false,       // 非阻塞
		nil,         // 额外参数
	)
	if err != nil {
		return fmt.Errorf("声明死信队列失败: %w", err)
	}

	// 绑定死信队列到死信交换机
	err = r.channel.QueueBind(
		r.deadQueue,    // 队列名称
		r.routingKey,   // 路由键
		r.deadExchange, // 交换机名称
		false,          // 非阻塞
		nil,            // 额外参数
	)
	if err != nil {
		return fmt.Errorf("绑定死信队列失败: %w", err)
	}

	// 声明正常交换机
	err = r.channel.ExchangeDeclare(
		r.exchange, // 交换机名称
		"direct",   // 交换机类型
		true,       // 持久化
		false,      // 自动删除
		false,      // 内部使用
		false,      // 非阻塞
		nil,        // 额外参数
	)
	if err != nil {
		return fmt.Errorf("声明交换机失败: %w", err)
	}

	// 声明正常队列，设置消息过期后转发到死信队列
	args := amqp.Table{
		"x-dead-letter-exchange":    r.deadExchange,
		"x-dead-letter-routing-key": r.routingKey,
		"x-message-ttl":             r.expiration * 1000, // 毫秒
	}

	_, err = r.channel.QueueDeclare(
		r.queueName, // 队列名称
		true,        // 持久化
		false,       // 自动删除
		false,       // 排他性
		false,       // 非阻塞
		args,        // 额外参数
	)
	if err != nil {
		return fmt.Errorf("声明队列失败: %w", err)
	}

	// 绑定队列到交换机
	err = r.channel.QueueBind(
		r.queueName,  // 队列名称
		r.routingKey, // 路由键
		r.exchange,   // 交换机名称
		false,        // 非阻塞
		nil,          // 额外参数
	)
	if err != nil {
		return fmt.Errorf("绑定队列失败: %w", err)
	}

	return nil
}

// PublishMessage 发布消息
func (r *RabbitMQ) PublishMessage(ctx context.Context, body []byte) error {
	// 设置消息持久化
	msg := amqp.Publishing{
		ContentType:  "application/json",
		Body:         body,
		DeliveryMode: amqp.Persistent, // 持久化
	}

	// 发布消息
	return r.channel.Publish(
		r.exchange,   // 交换机名称
		r.routingKey, // 路由键
		false,        // 强制发布
		false,        // 立即发布
		msg,          // 消息
	)
}

// ConsumeMessage 消费消息
func (r *RabbitMQ) ConsumeMessage(ctx context.Context, handler func([]byte) error) error {
	// 设置预取计数，控制消费速率
	err := r.channel.Qos(
		10,    // 预取计数
		0,     // 预取大小
		false, // 全局设置
	)
	if err != nil {
		return fmt.Errorf("设置QoS失败: %w", err)
	}

	// 获取消费通道
	msgs, err := r.channel.Consume(
		r.queueName, // 队列名称
		"",          // 消费者标签
		false,       // 自动确认
		false,       // 排他性
		false,       // 非本地
		false,       // 非阻塞
		nil,         // 额外参数
	)
	if err != nil {
		return fmt.Errorf("获取消费通道失败: %w", err)
	}

	// 处理消息
	go func() {
		for {
			select {
			case <-ctx.Done():
				// 上下文取消，退出goroutine
				log.Println("上下文取消，停止消息消费")
				return
			case msg, ok := <-msgs:
				if !ok {
					// 通道关闭，退出goroutine
					log.Println("消息通道已关闭，停止消息消费")
					return
				}
				// 处理消息
				err := handler(msg.Body)
				if err != nil {
					// 处理失败，拒绝消息并重新入队
					log.Printf("处理消息失败: %v, 消息将重新入队", err)
					msg.Nack(false, true)
				} else {
					// 处理成功，确认消息
					msg.Ack(false)
				}
			}
		}
	}()

	return nil
}

// ConsumeDeadLetterMessage 消费死信队列消息
func (r *RabbitMQ) ConsumeDeadLetterMessage(ctx context.Context, handler func([]byte) error) error {
	// 获取死信队列消费通道
	msgs, err := r.channel.Consume(
		r.deadQueue, // 队列名称
		"",          // 消费者标签
		false,       // 自动确认
		false,       // 排他性
		false,       // 非本地
		false,       // 非阻塞
		nil,         // 额外参数
	)
	if err != nil {
		return fmt.Errorf("获取死信队列消费通道失败: %w", err)
	}

	// 处理死信消息
	go func() {
		for {
			select {
			case <-ctx.Done():
				// 上下文取消，退出goroutine
				log.Println("上下文取消，停止死信消息消费")
				return
			case msg, ok := <-msgs:
				if !ok {
					// 通道关闭，退出goroutine
					log.Println("死信消息通道已关闭，停止消息消费")
					return
				}
				// 处理消息
				err := handler(msg.Body)
				if err != nil {
					// 处理失败，拒绝消息
					log.Printf("处理死信消息失败: %v", err)
					msg.Nack(false, false)
				} else {
					// 处理成功，确认消息
					msg.Ack(false)
				}
			}
		}
	}()

	return nil
}

// Close 关闭连接
func (r *RabbitMQ) Close() {
	if r.channel != nil {
		r.channel.Close()
	}
	if r.conn != nil {
		r.conn.Close()
	}
}

// ReconnectOnFailure 断线重连
func (r *RabbitMQ) ReconnectOnFailure() {
	// 默认超时时间，防止测试环境中无限运行
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	r.ReconnectOnFailureWithContext(ctx)
}

// ReconnectOnFailureWithContext 支持上下文的断线重连
func (r *RabbitMQ) ReconnectOnFailureWithContext(ctx context.Context) {
	// 监听连接关闭
	connCloseChan := make(chan *amqp.Error)
	r.conn.NotifyClose(connCloseChan)

	// 监听通道关闭
	channelCloseChan := make(chan *amqp.Error)
	r.channel.NotifyClose(channelCloseChan)

	// 处理连接断开
	go func() {
		for {
			select {
			case <-ctx.Done():
				// 上下文取消，退出goroutine
				log.Println("上下文取消，停止RabbitMQ重连")
				return
			case <-connCloseChan:
				log.Println("RabbitMQ连接已断开，尝试重新连接...")
			reconnectLoop:
				for {
					select {
					case <-ctx.Done():
						// 上下文取消，退出重连循环
						log.Println("上下文取消，停止RabbitMQ重连")
						return
					default:
						// 尝试重新连接
						newMQ, err := NewRabbitMQ(r.queueName, r.exchange, r.routingKey)
						if err != nil {
							log.Printf("重新连接RabbitMQ失败: %v, 5秒后重试", err)
							time.Sleep(5 * time.Second)
							continue
						}

						// 更新连接和通道
						r.conn = newMQ.conn
						r.channel = newMQ.channel

						// 重新初始化队列
						if err := r.InitOrderQueue(); err != nil {
							log.Printf("重新初始化队列失败: %v, 5秒后重试", err)
							time.Sleep(5 * time.Second)
							continue
						}
						break reconnectLoop
					}
				}
			case <-channelCloseChan:
				select {
				case <-ctx.Done():
					// 上下文取消，退出goroutine
					log.Println("上下文取消，停止RabbitMQ重连")
					return
				default:
					log.Println("RabbitMQ通道已断开，尝试重新连接...")
					time.Sleep(5 * time.Second)
				}
			}
		}
	}()
}

// DeclareExchange 声明交换机
func (r *RabbitMQ) DeclareExchange(exchangeName, exchangeType string) error {
	return r.channel.ExchangeDeclare(
		exchangeName, // 交换机名称
		exchangeType, // 交换机类型
		true,         // 持久化
		false,        // 自动删除
		false,        // 内部使用
		false,        // 非阻塞
		nil,          // 额外参数
	)
}

// DeclareQueue 声明队列
func (r *RabbitMQ) DeclareQueue(queueName string) error {
	_, err := r.channel.QueueDeclare(
		queueName, // 队列名称
		true,      // 持久化
		false,     // 自动删除
		false,     // 排他性
		false,     // 非阻塞
		nil,       // 额外参数
	)
	return err
}

// BindQueue 绑定队列到交换机
func (r *RabbitMQ) BindQueue(queueName, exchangeName, routingKey string) error {
	return r.channel.QueueBind(
		queueName,    // 队列名称
		routingKey,   // 路由键
		exchangeName, // 交换机名称
		false,        // 非阻塞
		nil,          // 额外参数
	)
}

// PublishWithContext 带上下文的消息发布
func (r *RabbitMQ) PublishWithContext(ctx context.Context, exchangeName, routingKey string, body []byte) error {
	// 设置消息持久化
	msg := amqp.Publishing{
		ContentType:  "application/json",
		Body:         body,
		DeliveryMode: amqp.Persistent, // 持久化
	}

	// 发布消息
	return r.channel.Publish(
		exchangeName, // 交换机名称
		routingKey,   // 路由键
		false,        // 强制发布
		false,        // 立即发布
		msg,          // 消息
	)
}

// Consume 消费消息并返回消息通道
func (r *RabbitMQ) Consume(queueName, consumerTag string) (<-chan amqp.Delivery, error) {
	// 获取消费通道
	return r.channel.Consume(
		queueName,   // 队列名称
		consumerTag, // 消费者标签
		false,       // 自动确认
		false,       // 排他性
		false,       // 非本地
		false,       // 非阻塞
		nil,         // 额外参数
	)
}
